package com.wsjj.yjh;


import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;

public class SinkKafkaWithKey {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.noRestart());

        DataStreamSource<String> source = env.socketTextStream("hadoop102", 7777);

        source.sinkTo(KafkaSink.<String>builder()
//                链接kafka
                        .setBootstrapServers("hadoop102:9092,hadoop103:9092")
//                事务前缀
                        .setTransactionalIdPrefix("yjh-")
//                精准消费
                        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
//                错误最大时长
                        .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,10 * 60 * 1000 + "")
//                自定义序列化
                        .setRecordSerializer(new KafkaRecordSerializationSchema<String>() {
                            @Nullable
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(String element, KafkaSinkContext context, Long timestamp) {
                                String[] split = element.split(",");
                                byte[] key = split[0].getBytes(StandardCharsets.UTF_8);
                                byte[] value = element.getBytes(StandardCharsets.UTF_8);
                                return new ProducerRecord<byte[], byte[]>("topic_sink",key,value);
                            }
                        })
                .build());
        env.execute();
    }
}
